package com.sensei.search.client;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;

import org.apache.log4j.Logger;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipelineCoverage;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;

import com.browseengine.bobo.api.BrowseRequest;
import com.browseengine.bobo.api.BrowseResult;
import com.browseengine.bobo.protobuf.BrowseProtobufConverter;
import com.browseengine.bobo.protobuf.BrowseRequestBPO;
import com.browseengine.bobo.protobuf.BrowseResultBPO;
import com.browseengine.bobo.protobuf.BrowseResultBPO.Result;

@ChannelPipelineCoverage("one")
public class SenseiClientHandler extends SimpleChannelUpstreamHandler {
  private static final Logger logger = Logger.getLogger(SenseiClientHandler.class);
  
  private volatile Channel _channel;
  private final BlockingDeque<BrowseResultBPO.Result> answer = new LinkedBlockingDeque<Result>();

  @Override
  public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e)
		throws Exception {
	_channel = e.getChannel();
	super.channelOpen(ctx, e);
  }
  
  public BrowseResult browse(BrowseRequest req){
	  BrowseRequestBPO.Request msg = BrowseProtobufConverter.convert(req);
	  _channel.write(msg);
	  BrowseResultBPO.Result resultMsg;
	  while(true){
		  try{
			  resultMsg = answer.take();
			  break;
		  }
		  catch(InterruptedException e){
			  continue;
		  }
	  }
	  
	  return BrowseProtobufConverter.convert(resultMsg);
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
		throws Exception {
	logger.error("Unexpected exception from downstream.",e.getCause());
	e.getChannel().close();
  }

  @Override
  public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
		throws Exception {
	boolean offered = answer.offer((Result)e.getMessage());
	assert offered;
  }
}
